package com.nepu.gmall.realtime.util;

import com.nepu.gmall.realtime.bean.TransientSink;
import com.nepu.gmall.realtime.common.GmallConfig;
import lombok.SneakyThrows;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author chenshuaijun
 * @create 2023-03-01 9:53
 */
public class ClickHouseUtil {

    public static <T> SinkFunction<T> getJdbcSink(String sql){

        return JdbcSink.sink(sql, new JdbcStatementBuilder<T>() {
            @SneakyThrows
            @Override
            public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                // 首先通过反射取出所有的属性
                Class<?> tcl = t.getClass();
                // 拿到所有的属性
                Field[] fields = tcl.getDeclaredFields();
                // 记录SQL中当前是第几个问号
                int  questionMarkNum = 1;
                // 遍历所有的属性
                for (int i = 0; i < fields.length; i++) {
                    Field field = fields[i];
                    field.setAccessible(true);
                    // 判断当前属性有没有被打上注解
                    TransientSink sinkAnnotation = field.getAnnotation(TransientSink.class);
                    if (sinkAnnotation != null){
                        // 表示的是当前的属性不作为填充的对象
                        continue;
                    }
                    //获得当前属性值
                    Object value = field.get(t);
                    // 将值进行填充
                    preparedStatement.setObject(questionMarkNum,value);
                    questionMarkNum++;
                }

            }
        },
                JdbcExecutionOptions
                        .builder()
                        .withBatchSize(5)
                        .withBatchIntervalMs(5000L)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .build()
        );
    }
}
